package com.shujia.flink.sql

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo2DataStreamToTable {
  def main(args: Array[String]): Unit = {

    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //使用blink计划器
      .inStreamingMode() //使用流模型
      .build()

    //创建flink sql的执行环境
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    val clicksDS: DataStream[String] = bsEnv.socketTextStream("master", 8888)


    /**
      * 用户编号，时间，url
      * 001,202111011003,baidu
      * 001,202111011003,baidu
      * 001,202111011003,baidu
      * 001,202111011003,baidu
      * 001,202111011003,baidu
      * 002,202111011003,baidu
      * 002,202111011003,baidu
      */

    val clickItemDS: DataStream[(String, String, String)] = clicksDS.map(line => {
      val split: Array[String] = line.split(",")
      val uid: String = split(0)
      val cTime: String = split(1)
      val url: String = split(2)

      (uid, cTime, url)
    })


    /**
      * 将流转换成表,指定字段名
      */

    val table: Table = bsTableEnv.fromDataStream(clickItemDS, $"uid", $"cTIme", $"url")

    table.printSchema()


    /**
      * DSL
      *
      */
    //输出表是一个更新的表
    /* val countTable: Table = table
       .groupBy($"uid")
       .select($"uid", $"url".count())*/


    /**
      * 创建临时视图
      *
      */

    //两种写法都一样
    //bsTableEnv.createTemporaryView("clicks", table)
    //将流直接注册成视图，写sql
    bsTableEnv.createTemporaryView("clicks", clickItemDS, $"uid", $"cTIme", $"url")

    /**
      * SQL
      *
      * executeSql: 可以直接所有的sql语句（DDL ,DSL, DQL）,  不会返回sql执行的查询结果
      * sqlQuery； 执行查询的sql ，会sql查询的结果
      */

    val resultTable: Table = bsTableEnv.sqlQuery(
      """
        |select uid,count(url) from clicks group by uid
        |
      """.stripMargin)


    /**
      * 将表转换成流,
      * 更新的流的key 是插入删除的标记， value是一行数据
      *
      */
    val countDS: DataStream[(Boolean, Row)] = resultTable.toRetractStream[Row]

    countDS.print()

    bsEnv.execute()

  }

}
